#include "raft_log.h"

#include "zrpc/base/logger.h"
#include "zrpc/net/socket.h"
#include "raft/raft.h"

namespace libraft {
RaftLog::RaftLog(const std::shared_ptr<rocksdb::DB> &db)
    : log_(db), last_log_index_(0), first_log_index_(0) {
  std::unique_ptr<rocksdb::Iterator> it(
      log_->NewIterator(rocksdb::ReadOptions()));
  it->SeekToFirst();
  if (it->Valid()) {
    first_log_index_ = zrpc::socket::NetworkToHost64(
        slash::DecodeFixed64(it->key().ToString().c_str()));
  }

  it->SeekToLast();
  if (it->Valid()) {
    last_log_index_ = zrpc::socket::NetworkToHost64(
        slash::DecodeFixed64(it->key().ToString().c_str()));
  }
}

RaftLog::~RaftLog() {}

uint64_t RaftLog::Append(const raft::Entry &entry) {
  std::unique_lock<std::mutex> lk(mutex_);
  std::string buf;
  assert(entry.SerializeToString(&buf));
  last_log_index_++;
  std::string last_log_index_string;
  slash::PutFixed64(&last_log_index_string,
                    zrpc::socket::HostToNetwork64(last_log_index_));
  rocksdb::Status s =
      log_->Put(rocksdb::WriteOptions(), last_log_index_string, buf);
  assert(s.ok());

  if (!s.ok()) {
    last_log_index_--;
  }
  return last_log_index_;
}

uint64_t RaftLog::AppendMulti(const std::vector<const raft::Entry *> &entries) {
  std::unique_lock<std::mutex> lk(mutex_);
  rocksdb::WriteBatch write_batch;
  for (size_t i = 0; i < entries.size(); i++) {
    std::string buf;
    assert(entries[i]->SerializeToString(&buf));
    last_log_index_++;

    std::string last_log_index_string;
    slash::PutFixed64(&last_log_index_string,
                      zrpc::socket::HostToNetwork64(last_log_index_));
    assert(write_batch.Put(last_log_index_string, buf).ok());
  }

  rocksdb::Status s = log_->Write(rocksdb::WriteOptions(), &write_batch);
  assert(s.ok());

  if (!s.ok()) {
    last_log_index_ -= entries.size();
  }
  return last_log_index_;
}

bool RaftLog::MultiGetEntry(const std::vector<uint64_t> &indexes,
                            std::vector<raft::Entry *> &entries) {
  std::unique_lock<std::mutex> lk(mutex_);
  std::vector<rocksdb::Slice> index_strs;
  for (int i = 0; i < indexes.size(); i++) {
    std::string index_string;
    slash::PutFixed64(&index_string, zrpc::socket::HostToNetwork64(indexes[i]));
    index_strs.push_back(index_string);
  }

  std::vector<std::string> values;
  std::vector<rocksdb::Status> s =
      log_->MultiGet(rocksdb::ReadOptions(), index_strs, &values);
  for (int i = 0; i < s.size(); i++) {
    if (s[i].ok()) {
      assert(entries[i]->ParseFromString(values[i]));
    } else {
      entries[i] = nullptr;
      return false;
    }
  }
  return true;
}

bool RaftLog::GetEntry(uint64_t index, raft::Entry *entry) {
  std::unique_lock<std::mutex> lk(mutex_);
  std::string index_string;
  slash::PutFixed64(&index_string, zrpc::socket::HostToNetwork64(index));

  std::string entry_buf;
  rocksdb::Status s =
      log_->Get(rocksdb::ReadOptions(), index_string, &entry_buf);
  if (s.ok()) {
    if (entry->ParseFromString(entry_buf)) {
      return true;
    } else {
      return false;
    }
  } else {
    return false;
  }
}

uint64_t RaftLog::LastLogIndex() { return last_log_index_; }

uint64_t RaftLog::FirstLogIndex() { return first_log_index_; }

bool RaftLog::GetLastLogTermAndIndex(uint64_t *last_log_term,
                                     uint64_t *last_log_index) {
  std::unique_lock<std::mutex> lk(mutex_);
  if (last_log_index_ == 0) {
    *last_log_index = 0;
    *last_log_term = 0;
    return true;
  }

  std::string buf;
  std::string last_log_index_string;
  slash::PutFixed64(&last_log_index_string,
                    zrpc::socket::HostToNetwork64(last_log_index_));
  rocksdb::Status s =
      log_->Get(rocksdb::ReadOptions(), last_log_index_string, &buf);
  if (!s.ok() || s.IsNotFound()) {
    *last_log_index = 0;
    *last_log_term = 0;
    return true;
  }

  raft::Entry entry;
  assert(entry.ParseFromString(buf));
  *last_log_index = last_log_index_;
  *last_log_term = entry.term();
  return true;
}

rocksdb::Status RaftLog::TruncateSuffix(uint64_t index) {
  rocksdb::WriteBatch write_batch;
  for (; last_log_index_ >= index; last_log_index_--) {
    std::string last_log_index_string;
    slash::PutFixed64(&last_log_index_string,
                      zrpc::socket::HostToNetwork64(last_log_index_));
    write_batch.Delete(last_log_index_string);
  }

  if (write_batch.Count() > 0) {
    return log_->Write(rocksdb::WriteOptions(), &write_batch);
  }
  return rocksdb::Status::OK();
}

rocksdb::Status RaftLog::TruncatePrefix(uint64_t index) {
  std::string prefix_log_index_string;
  slash::PutFixed64(&prefix_log_index_string, zrpc::socket::HostToNetwork64(0));

  std::string log_index_string;
  slash::PutFixed64(&log_index_string, zrpc::socket::HostToNetwork64(index));
  return log_->DeleteRange(rocksdb::WriteOptions(), nullptr,
                           prefix_log_index_string, log_index_string);
}

}  // namespace libraft